package com.lnj.user.service.Impl;

import co.elastic.clients.elasticsearch.ElasticsearchClient;
import co.elastic.clients.elasticsearch.core.*;
import co.elastic.clients.elasticsearch.core.search.Hit;
import com.baomidou.mybatisplus.core.conditions.query.QueryWrapper;
import com.lnj.api.FileUploadApi;
import com.lnj.api.IdGeneratorApi;
import com.lnj.bean.dto.UserDTO;
import com.lnj.bean.model.User;
import com.lnj.user.mapper.UserMapper;
import com.lnj.user.service.UserService;
import com.lnj.user.utils.*;
import com.lnj.utils.ResponseResult;
import lombok.extern.java.Log;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.security.crypto.password.PasswordEncoder;
import org.springframework.stereotype.Service;
import org.springframework.transaction.annotation.Transactional;
import org.springframework.web.multipart.MultipartFile;

import java.util.*;
import java.util.stream.Collectors;

@Service
@Log
public class UserServiceImpl implements UserService {
    private final UserMapper userMapper; // 注入 UserMapper
    private final PasswordEncoder passwordEncoder;   //spring security提供的密码的加密器
    private final JwtTokenUtil jwtTokenUtil;    // 生成和验证token的工具类

    @Autowired
    private IdGeneratorApi idGeneratorApi;
    @Autowired
    private FileUploadApi fileUploadApi;
    @Autowired
    private ElasticsearchClient elasticsearchClient;
    @Autowired
    private KafkaProducterUtil kafkaProducterUtil;
    @Autowired
    private JsonUtil jsonUtil;


    public UserServiceImpl(UserMapper userMapper, PasswordEncoder passwordEncoder, JwtTokenUtil jwtTokenUtil) {
        this.userMapper = userMapper;
        this.passwordEncoder = passwordEncoder;
        this.jwtTokenUtil = jwtTokenUtil;
    }

    /**
     * 用户注册。
     * @param userDTO 注册信息
     * @return 注册成功的用户DTO
     * @throws IllegalArgumentException 如果用户名已存在
     */
    @Transactional
    public UserDTO registerUser(UserDTO userDTO) {
        if (checkUsernameExists(userDTO.getUserName())) { // 使用新的 checkUsernameExists
            throw new IllegalArgumentException("Username already exists: " + userDTO.getUserName());
        }

        //通过 feign客户端获得用户id
        ResponseResult rr = this.idGeneratorApi.getNextId();
        if (userDTO.getAvatarFile() != null) {
            ResponseResult rr2 = this.fileUploadApi.upload(new MultipartFile[]{userDTO.getAvatarFile()});
            if (rr2.getCode() == 0) {
                // 部分成功 - data是拼接的字符串
                String resultData = rr2.getData().toString();
                // 解析出URL部分（假设格式为"url,url失败:xxx"）
                String[] parts = resultData.split("失败:");
                if (parts.length > 0 && !parts[0].isEmpty()) {
                    String[] urls = parts[0].split(",");
                    if (urls.length > 0) {
                        userDTO.setAvatar(urls[0]); // 取第一个URL
                    }
                }
                // 抛出异常包含失败信息
                throw new RuntimeException("图片上传部分失败: " + (parts.length > 1 ? parts[1] : "未知原因"));
            } else if (rr2.getCode()==2){
                // 全部失败 - data是失败信息字符串
                throw new RuntimeException("图片上传失败: " + rr2.getData());
            }
            List<String> avatarUrls = (List<String>) rr2.getData();
            userDTO.setAvatar(avatarUrls.get(0));
        }

        User user = new User();
        user.setUserId(  Long.parseLong( rr.getData().toString()) );
        user.setUserName(userDTO.getUserName());
        user.setUserPassword(     passwordEncoder.encode(userDTO.getUserPassword())   ); // 密码加密存储
        user.setMobile(userDTO.getMobile());
        user.setEmail(userDTO.getEmail());
        user.setGender(userDTO.getGender());
        user.setUserStatus(1);
        // 默认注册为普通用户，角色存储为逗号分隔的字符串
        user.setRoles("CUST"); // 默认角色
        user.setAvatar(userDTO.getAvatar());

        userMapper.insert(user); // 使用 MyBatis Plus 的 insert 方法
        UserDTO userDTO1=convertToDto(user);
        //TODO: 消息队列
        String json = jsonUtil.toJson(userDTO1);
        kafkaProducterUtil.sendMessage("es_data_topic","users",json);
        kafkaProducterUtil.sendMessage("email_topic","user_register",json);
        setES(userDTO1);
        return userDTO1; // 返回转换后的DTO
    }

    /**
     * 用户登录。
     * @param authRequest 登录请求（用户名和密码）
     * @return 包含 JWT Token 和用户信息的响应
     * @throws IllegalArgumentException 如果用户名或密码不正确
     */
    public AuthResponse loginUser(AuthRequest authRequest) {
        QueryWrapper<User> queryWrapper = new QueryWrapper<>();
        queryWrapper.eq("userName", authRequest.getUserName());
        User user = userMapper.selectOne(queryWrapper); // 使用 MyBatis Plus 的 selectOne

        if (user == null) {
            throw new IllegalArgumentException("Invalid username or password.");
        }

        if (!passwordEncoder.matches(authRequest.getUserPassword(), user.getUserPassword())) {
            throw new IllegalArgumentException("Invalid username or password.");
        }

        // 登录成功，生成 JWT Token
        // 将逗号分隔的角色字符串转换为 Set<String> 传递给 JWT 工具
        Set<String> rolesSet = Arrays.stream(user.getRoles().split(","))
                .map(String::trim)
                .collect(Collectors.toSet());

        String token = jwtTokenUtil.generateToken(user.getUserId(), user.getUserName(), rolesSet);

        //发送Kafka消息去推荐服务
        kafkaProducterUtil.sendMessage("recommend_topic",user.getUserId().toString());

        return new AuthResponse(token, convertToDto(user));
    }

    /**
     * 判断用户名是否存在（ES优先查询）
     * @param username 用户名
     * @return 如果存在返回 true，否则返回 false
     */
    public boolean checkUsernameExists(String username) {
        try {
            // 1. 先查ES - 使用新API
          SearchResponse<UserDTO> response = elasticsearchClient.search(SearchRequest.of(s -> s
                            .index("users")
                            .query(q -> q
                                    .term(t -> t
                                            .field("userName.keyword")
                                            .value(username)
                                    )
                            )
                            .size(0)), // 不返回实际文档
                    UserDTO.class);

            if (response.hits().total().value() > 0) {
                return true;
            }

            // 2. ES不存在则降级查数据库
            return checkUsernameInDB(username);
        } catch (Exception e) {
            log.warning("ES查询异常，降级到数据库查询");
            return checkUsernameInDB(username);
        }
    }

    /**
     * 根据用户ID查询用户信息（ES优先）
     * @param id 用户ID
     * @return 用户DTO
     */
    public Optional<UserDTO> getUserById(Long id) {
        try {
            // 1. 先查ES - 使用新API
            GetResponse<UserDTO> response = elasticsearchClient.get(GetRequest.of(g -> g
                            .index("users")
                            .id(id.toString())),
                    UserDTO.class);

            if (response.found()) {
                return Optional.ofNullable(response.source());
            }

            // 2. ES不存在则降级查数据库
            return fallbackGetUserFromDB(id);
        } catch (Exception e) {
            log.warning("ES查询异常，降级到数据库查询");
            return fallbackGetUserFromDB(id);
        }
    }

    /**
     * 获取所有用户信息（ES优先）
     * @return 用户DTO列表
     */
    public List<UserDTO> getAllUsers() {
        try {
            // 1. 先查ES - 使用新API
            SearchResponse<UserDTO> response = elasticsearchClient.search(SearchRequest.of(s -> s
                            .index("users")
                            .query(q -> q.matchAll(m -> m))
                            .size(10000)), // 设置合理上限
                    UserDTO.class);

            return response.hits().hits().stream()
                    .map(Hit::source)
                    .collect(Collectors.toList());
        } catch (Exception e) {
            log.warning("ES查询异常，降级到数据库查询");
            return fallbackGetAllUsersFromDB();
        }
    }

    // 数据库降级查询方法保持不变
    private boolean checkUsernameInDB(String username) {
        QueryWrapper<User> queryWrapper = new QueryWrapper<>();
        queryWrapper.eq("userName", username);
        return userMapper.exists(queryWrapper);
    }

    private Optional<UserDTO> fallbackGetUserFromDB(Long id) {
        User user = userMapper.selectById(id);
        return Optional.ofNullable(user).map(this::convertToDto);
    }

    private List<UserDTO> fallbackGetAllUsersFromDB() {
        return userMapper.selectList(null).stream()
                .map(this::convertToDto)
                .collect(Collectors.toList());
    }

    /**
     * 更新用户信息。
     * @param id 用户ID
     * @param userDTO 包含更新信息的DTO
     * @return 更新后的用户DTO
     * @throws IllegalArgumentException 如果用户不存在
     */
    @Transactional
    public UserDTO updateUser(Long id, UserDTO userDTO) {
        User existingUser = userMapper.selectById(id); // 使用 MyBatis Plus 的 selectById
        if (existingUser == null) {
            throw new IllegalArgumentException("User not found with ID: " + id);
        }

        // 允许更新用户名和角色，但通常不直接更新密码（密码有单独的修改接口）
        if (userDTO.getUserName() != null && !userDTO.getUserName().equals(existingUser.getUserName())) {
            if (checkUsernameExists(userDTO.getUserName())) { // 使用新的 checkUsernameExists
                throw new IllegalArgumentException("New username already exists: " + userDTO.getUserName());
            }
            existingUser.setUserName(userDTO.getUserName());
        }

        // 角色更新（通常只有管理员能修改角色）
        if (userDTO.getRoles() != null && !userDTO.getRoles().isEmpty()) {
            // 将 Set<String> 转换为逗号分隔的字符串
            existingUser.setRoles(String.join(",", userDTO.getRoles()));
        }


        int i = userMapper.updateById(existingUser);// 使用 MyBatis Plus 的 updateById
        if(i<=0){
            throw new IllegalArgumentException("更新用户失败");
        }
        UserDTO userDTO1 = convertToDto(existingUser);
        //TODO: 消息队列
        String json = jsonUtil.toJson(userDTO1);
        kafkaProducterUtil.sendMessage("es_data_topic","users",json);
        return userDTO1; // 返回转换后的DTO
    }

    /**
     * 用户注销（通常是客户端删除 Token，服务端无状态）。
     * 如果需要黑名单机制，可以在此实现。
     * @param token 待注销的 token
     */
    public boolean logout(String token) {
        if (token != null) {
            jwtTokenUtil.addToBlacklist(token);
           return true;
        } else {
            return false;
        }
    }

    /**
     * 将 User 实体转换为 UserDTO。
     */
    private UserDTO convertToDto(User user) {
        UserDTO dto = new UserDTO();
        dto.setUserId(user.getUserId());
        dto.setUserName(user.getUserName());
        // 将逗号分隔的角色字符串转换为 Set<String>
        if (user.getRoles() != null && !user.getRoles().isEmpty()) {
            dto.setRoles(Arrays.stream(user.getRoles().split(","))
                    .map(String::trim)
                    .collect(Collectors.toSet()));
        } else {
            dto.setRoles(Collections.emptySet());
        }
        dto.setMobile(user.getMobile());
        dto.setEmail(user.getEmail());
        dto.setGender(user.getGender());
        dto.setAvatar(user.getAvatar());
        dto.setUserStatus(user.getUserStatus());
        // 不返回密码
        return dto;
    }

    /**
     * 计算路由 用来根据随机数计算用户ES存储的路由
     * @return
     */
    private String calculateRouting() {
        Random random = new Random();
        int shardIndex = random.nextInt(10);
        return "shard_" + shardIndex;
    }

    private void setES(UserDTO userDTO) {
        try {
            IndexRequest<UserDTO> request = IndexRequest.of(b -> b
                    .index("users")
                    .id(userDTO.getUserId().toString())
                    .routing(calculateRouting())
                    .document(userDTO)
            );
            elasticsearchClient.index(request);
        } catch (Exception esEx) {
            log.warning("用户同步到ES失败: "+userDTO.getUserId());
            // 这里可以选择记录失败日志或加入重试队列，但不影响主流程
        }
    }
}
